// ==================================================================================
// Shared Genomics Project MPI Codebase
// Version 1.0 30/04/2010
//
// (c) 2010 University of Manchester all rights reserved
//
// This file is distributed under the GNU General Public License, Version 2.  
// Please see the file COPYING.txt for more details
// ==================================================================================

/*!
\file
\brief	Main Body of the FAST_PEPI_BT_MERGE MPI Program
*/
/*! 
\defgroup	fast_pepi_bt_merge FAST_PEPI_BT_MERGE - Permutation Results Merge Program
\ingroup    mpi_stats
\brief		FAST_PEPI_BT_MERGE - Permutation Results Merge Program
\deprecated Program replaced by \ref fast_pepi_bt_v2 which combines the jobs of statistical calculation and result merging.
\details
<p>
  This program is based on the fast epistasis (BT) %option of PLINK.
  It performs the same interaction calculation and unlike PLINK can also permute interaction calculations by distributing
  randomised data-sets across the cores of a computer cluster.
</p>

<p>This program is used in conjunction with \ref fast_pepi_bt_merge and performs a permuted epistasis calculation which 
calculates the False Discovery Rates (FDR).</p>

<p>By setting an epistasis p-value cut-off threshold, this application will track interactions of interest in the permuted calculation.
Only interactions below the cut-off threshold are written to the results file.
The program has been tested, calculating the interactions of set up to 5000 SNPs.</p>

<h3>Usage</h3>
<p>The sequence of calls between the calculation and merge programs of fast epistasis (BT) are as follows:-</p>
<ul>
<li><strong>Call 1:-</strong><br>
fast_pepi_bt.exe performs a non-permuted interaction calculations in a data-set
which is spread across N-cores.</li>
<li><strong>Call 2:-</strong><br>
fast_pepi_bt_merge.exe (single-core) combines the interaction data from all the cores into a single
data file and identifies any interactions of interest based on a p-value significance threshold.</li>
<li><strong>Call 3:-</strong><br>
fast_pepi_bt.exe runs permutations on a randomised data-sets, tracking interactions of interest 
and writing permutation counts to file. The number of required permutations is spread over N-cores.
</li>
<li><strong>Call 4:-</strong><br>
fast_pepi_bt_merge.exe reads the permutation data from interesting interactions, 
combines this data and generates the final output file.</li>
</ul>

<p>If successful FAST_PEPI_BT_MERGE returns with an exitcode of 0, if it encountered an error the programs returns with an exit code of -1.<br>
This program creates output files containing statistical data about interesting interactions with the file suffix \ref SUFFIX_FEPI_FILE.
</p>

<p>This documentation is vague. To get a handle on how to use this code, look at the relevant code snippet in the \ref examples
application which outlines the invocation of this epistasis program.</p>

<h2>OPTIONS</H2>

<table class=options border=0>
<tr><th class=param_header colspan=1>PARAMETER</th>
<th class=param_header colspan=3>DESCRIPTION</th></tr>

<tr><td class=param_name colspan=1>--id STRING</td>
<td class=param_desc colspan=3>Unique identifer for a cluster job (also sets output file stem).</td></tr>

<tr><td class=param_name colspan=1>--rdir PATH</td>
<td class=param_desc colspan=3>Path to remote working directory on central file system.</td></tr>

<tr><td class=param_name colspan=1>--ldir PATH</td>
<td class=param_desc colspan=3>Path to local working directory on each core of the cluster.</td></tr>

<tr><td class=param_name colspan=1>--bmap FILE</td>
<td class=param_desc colspan=3>Path to the BMAP file (binary-encoded SNP list).</td></tr>

<tr><td class=param_name colspan=1>--permuted</td>
<td class=param_desc colspan=3>If this status flag is set, program attempts to read all of the permutation
count data and writes the final output file.</td></tr>

</table>
*/

#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include "adjusted.h"
#include "bmapfile.h"
#include "copyfile.h"
#include "doublefile.h"
#include "getopt.h"
#include "_int2.h"
#include "mt19937ar.h"
#include "options.h"
#include "pmaths.h"
#include "random_sleep.h"
#include "sort.h"
#include "suffixes.h"
#include "types.h"
#include "utils.h"

/*!
\brief Command-line options of FAST_PEPI_BT_MERGE
*/
struct option long_options[] = {
	{ "id", required_argument, NULL, 'i' }, // Run ID
	{ "bmap", required_argument, NULL, 'm' }, // Path to the map file.
	{ "rdir", required_argument, NULL, 'r'}, // Remote Output direct
	{ "ldir", required_argument, NULL, 'l'}, // Local Output direct
	{ "permuted", optional_argument, NULL, 'p'}, // Flag if integrating permuted data.
	{ 0,    0,    0,    0} 
};

#ifndef MAX_LINE_LENGTH
#define MAX_LINE_LENGTH 1024
#endif

/*!
\brief Main body of the FAST_PEPI_BT_MERGE program
\ingroup fast_pepi_bt_merge
\deprecated Program replaced by \ref fast_pepi_bt_v2 which combines the jobs of statistical calculation and result merging.
*/
int fast_pepi_merge_main(int argc, char *argv[]) {
	char ch;
	struct selected_options ops;
	BOOL jobDone = FALSE;
	int nRequiredArgs = 4, i, numprocs = -1;
	char localFilepath[FILE_PATH_LENGTH], remoteFilepath[FILE_PATH_LENGTH], line[MAX_LINE_LENGTH];
	char localFilepath2[FILE_PATH_LENGTH], token[80];
	BOOL isPermuted = FALSE;
	
	options_init(&ops);
	init_genrand((unsigned long) time(NULL));

	while ((ch = (char) getopt_long(argc, argv, "m:v:p:f:i:", long_options, NULL)) != -1) {
		switch (ch) {
			case 'i':
				strcpy(ops.szRunId, optarg);
				nRequiredArgs--;
				break;
			case 'm': // Bmap File
				strcpy(ops.szBmapFileName, optarg);
				nRequiredArgs--;
				break;
			case 'r': // Remote output directory
				strcpy(ops.rdir, optarg);
				nRequiredArgs--;
				break;
			case 'l': // Local output directory
				strcpy(ops.ldir, optarg);
				nRequiredArgs--;
				break;
			case 'p': // Dealing with permuted data.
				isPermuted = TRUE;
				break;
			default:
				break;
		}
	}

	// If the required arguements are not specified, there's no point carrying on.
	if (nRequiredArgs != 0) return -1;

	// Data merge in 2 modes, integration of the permuted and non-randomised result-sets.
	if (!isPermuted) {
		BOOL *epi_file_failures = NULL, *bepi_file_failures = NULL;
		int nInteractions = 0, nMap = 0;
		BOOL allEpiFilesCopied = TRUE, copiedSingleEpiFile = FALSE, createdRemoteEpiFile = FALSE, createdMergeLog = FALSE;
		BOOL allBepiFilesCopied = TRUE, copiedSingleBepiFile = FALSE, createdRemoteBepiFile = FALSE;
		
		numprocs = readNumberOfProcs(remoteFilepath, line, MAX_LINE_LENGTH, ops.rdir, ops.szRunId, SUFFIX_LOCAL_EPI_FILE); // The EPI file.
		if (numprocs < 1) return -1;

		epi_file_failures = create_bool_array(numprocs, FALSE); // Significant interaction file.
		bepi_file_failures = create_bool_array(numprocs, FALSE); // Best partner interaction file.
		nMap = bmapfile_read_nSNPs(&ops);
		
		// Start catting the separate outputs into a single file.
		if (nMap > 0 && epi_file_failures != NULL && bepi_file_failures != NULL) {
			// Copy the output file to the local node.
			for (i = 0; i < numprocs; i++) {
				set_datafile_path(remoteFilepath, ops.rdir, ops.szRunId, i, SUFFIX_LOCAL_EPI_FILE);
				set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_LOCAL_EPI_FILE);

				// Copy the remote output file from central server to local node
				// for file catting.
				if (copyFile(remoteFilepath, localFilepath)) copiedSingleEpiFile = TRUE;
				else {
					epi_file_failures[i] = TRUE;
					allEpiFilesCopied = FALSE;
				} 
			}

			// Join all the local EPI files together but only if
			// the copy process was successful across all nodes.
			if (allEpiFilesCopied) {
				FILE *out = NULL, *in = NULL;

				sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_EPI_FILE);
				out = fopen(localFilepath, "w");
				if (out != NULL) {
					// Asssume once the file is copied locally, its available to read without any hassle.
					for (i = 0; i < numprocs; i++) {
						set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_LOCAL_EPI_FILE);

						in = fopen(localFilepath, "r");
						if (in != NULL) {
							fgets(line, MAX_LINE_LENGTH, in); // Skip the header line.
							while(fgets(line, MAX_LINE_LENGTH, in) != NULL) {
								if (strlen(line) > 0) {
									fprintf(out,"%i,%s", nInteractions++, line);
								}
							}
							fclose(in);
						} 
					}
					fclose(out);

					// Create a binary file containing the pair-wise interactions.
					sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_EPI_FILE);
					in = fopen(localFilepath, "r");
					if (in != NULL) {
						sprintf(localFilepath2, "%s_%s%s", ops.ldir, ops.szRunId, SUFFIX_EPI_FILE2);
						out = fopen(localFilepath2, "wb");
						if (out != NULL) {
							struct INT2 i2;

							int2_init(&i2);
							i = 0;
							if(fwrite(&nInteractions, sizeof(int), 1, out) == 1) {
								while(fgets(line, MAX_LINE_LENGTH, in) != NULL && i < nInteractions) {
									if (strlen(line) > 0) {
										get_token(line, token, 0, ',');
										get_token(line, token, 1, ',');
										i2.p1 = atoi(token);
										get_token(line, token, 2, ',');
										i2.p2 = atoi(token);
										get_token(line, token, 3, ',');
										i2.z = atof(token);
										
										if(fwrite(&i2, sizeof(struct INT2), 1, out) == 1) i++;
									}
								}
							}
							fclose(out);
							if (nInteractions == i) createdRemoteEpiFile = TRUE;
						}
						fclose(in);
					}
					// Delete Text EPI file.
					sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_EPI_FILE);
					deleteFile(localFilepath);

					// Copy the local EPI file to remote directory path
					if (createdRemoteEpiFile) {
						sprintf(localFilepath, "%s_%s%s", ops.ldir, ops.szRunId, SUFFIX_EPI_FILE2);
						sprintf(remoteFilepath, "%s%s%s", ops.rdir, ops.szRunId, SUFFIX_EPI_FILE2);
						createdRemoteEpiFile = copyAndDeleteFile(localFilepath, remoteFilepath) ? TRUE : FALSE;
					}
				}
			}

			// Only bother integrating the best partner file if the remote EPI file
			// created on the central file server.
			if (createdRemoteEpiFile) {
				for (i = 0; i < numprocs; i++) {
					set_datafile_path(remoteFilepath, ops.rdir, ops.szRunId, i, SUFFIX_LOCAL_BEPI_FILE);
					set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_LOCAL_BEPI_FILE);

					if (copyFile(remoteFilepath, localFilepath)) copiedSingleBepiFile = TRUE;
					else {
						bepi_file_failures[i] = TRUE;
						allBepiFilesCopied = FALSE;
					} 
				}

				// Write the contents of the best partner file to the local output file.
				if (allBepiFilesCopied) {
					int *summary_sig = NULL, *summary_good = NULL, *best_partner = NULL;
					double *best_score = NULL;
					int filesParsed = 0;

					// Allocate the result arrays for the pairwise interactions.
					summary_sig = create_int_array(nMap, 0);
					summary_good = create_int_array(nMap, 0);
					best_partner = create_int_array(nMap, -1);
					best_score = create_vector(nMap, 0);

					if (summary_sig != NULL && summary_good != NULL && best_partner != NULL && best_score != NULL) {
						// Parse each BEPI file and combine the results into the output array.
						for (i = 0; i < numprocs; i++) {
							FILE *in = NULL;

							set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_LOCAL_BEPI_FILE);
							in = fopen(localFilepath, "r");
							if (in != NULL) {
								while(fgets(line, MAX_LINE_LENGTH, in) != NULL) {
									if (strlen(line) > 0) {
										char *p, *q;
										char token[80];
										int tokenCount = 0, snpIndex = 0, _summary_sig = 0, _summary_good = 0, _best_partner = -1;
										double _best_score = -1;

										// Read a token at a time from the line
										p = line;
										while (*p) {
											q = token;
											while (*p != ',' && *p != '\r' && *p != '\n' && *p) {
												*q = *p;
												q++;
												p++;
											}
											if (*p) p++; // Next token
											*q = '\0'; // Null terminate token.

											// Convert the strings to numbers and update the SNP
											// epistasis result array.
											switch (tokenCount) {
												case 0:
													snpIndex = atoi(token);
													break;
												case 1:
													_summary_sig = atoi(token);
													break;
												case 2:
													_summary_good = atoi(token);
													break;
												case 3:
													_best_score = atof(token);
													break;
												case 4:
													_best_partner =  atoi(token);
													break;
												default:
													break;
											}
											tokenCount++;
										}

										// Update the result arrays
										summary_sig[snpIndex] += _summary_sig;
										summary_good[snpIndex] += _summary_good;
										if (_best_score > best_score[snpIndex]) {
											best_score[snpIndex] = _best_score;
											best_partner[snpIndex] = _best_partner;
										}
									}
								}
								fclose(in);
								filesParsed++;
							}
						}

						// Write the results arrays to an output merged BEPI file
						// and then copy over to the central server once all done.
						if (filesParsed == numprocs) {
							FILE *out = NULL;
							sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_BEPI_FILE);
							sprintf(remoteFilepath, "%s%s%s", ops.rdir, ops.szRunId, SUFFIX_BEPI_FILE);

							out = fopen(localFilepath, "w");
							if (out != NULL) {
								fprintf(out, "SNP,N_SIG,N_TOT,BEST_CHISQ,BEST_SNP\n");

								// Write out the best pair-wise results for each SNP.
								for (i = 0; i < nMap; i++) {
									fprintf(out, "%i,%i,%i,%f,", i, summary_sig[i], summary_good[i], best_score[i] * best_score[i]);
									if (best_partner[i] != -1) 
										fprintf(out, "%i\n", best_partner[i]);
									else 
										fprintf(out, "Na\n");
								}
								fclose(out);

								// Copy the combined BEPI file to the remote directory path.
								if (copyAndDeleteFile(localFilepath, remoteFilepath)) {
									if(doublefile_write(best_score, nMap, &ops, SUFFIX_DOUBLE_FILE)) createdRemoteBepiFile = TRUE;
								}
							}
						}
					}
					// Free the result arrays.
					if (summary_sig != NULL) free(summary_sig);
					if (summary_good != NULL) free(summary_good);
					if (best_partner != NULL) free(best_partner);
					if (best_score != NULL) free(best_score);
				}
			}

			if (copiedSingleEpiFile) {
				FILE *out = NULL;

				sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_MERGE_LOG);
				sprintf(remoteFilepath, "%s%s%s", ops.rdir, ops.szRunId, SUFFIX_MERGE_LOG);	
				out = fopen(localFilepath, "w");
				if (out != NULL) {
					time_t now = time(NULL);

					fprintf(out, "Dated=%s", ctime(&now));
					fprintf(out, "Interactions=%i\n", nInteractions);
					fprintf(out, "Numprocs=%i\n", numprocs);

					// If a node failed, write out failed ranks to the output file.
					if (!allEpiFilesCopied) {
						fprintf(out, "Failed Ranks=");
						for (i = 0; i < numprocs; i++) {
							if (epi_file_failures[i] == TRUE) {
								fprintf(out, "%i ", i);
								bepi_file_failures[i] = FALSE; // Set to false to avoid repeated numbers in the failed rank section of the merge log
							} 
							if (bepi_file_failures[i] == TRUE) {
								fprintf(out, "%i ", i);
							}
						}
						fprintf(out, "\n");
					}
					fclose(out);

					// Copy the combined BEPI file to the remote directory path.
					if (copyAndDeleteFile(localFilepath, remoteFilepath)) createdMergeLog = TRUE;
				}
			}
		}
		if (epi_file_failures != NULL) free(epi_file_failures);
		if (bepi_file_failures != NULL) free(bepi_file_failures);
		
		// Delete the temporary data files on the local node.
		if (copiedSingleEpiFile) {
			for (i = 0; i < numprocs; i++) {
				set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_LOCAL_EPI_FILE);
				deleteFile(localFilepath);

				set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_LOCAL_BEPI_FILE);
				deleteFile(localFilepath);
			}
		}

		// If all is done, flag eveything's tickety-boo.
		if (createdRemoteEpiFile && createdRemoteBepiFile && createdMergeLog) {
			jobDone = TRUE;

			// Delete the temp output files from the remote directory.
			for (i = 0; i < numprocs; i++) {
				set_datafile_path(remoteFilepath, ops.rdir, ops.szRunId, i, SUFFIX_LOCAL_EPI_FILE);
				deleteFile(remoteFilepath);
				set_datafile_path(remoteFilepath, ops.rdir, ops.szRunId, i, SUFFIX_LOCAL_BEPI_FILE);
				deleteFile(remoteFilepath);
			}
		}
	} else { 
		// Join up the permutated data counts.
		int nSNPs = 0, nPermutations = 0, *bp_perm_counts = NULL;
		int nInterestingInteractions = 0, *perms_done = NULL;
		struct INT2 *interesting_interactions = NULL;
		struct small_locus *map = bmapfile_load(&ops, &nSNPs);
		
		if (nSNPs > 0 && map != NULL) bp_perm_counts = create_int_array(nSNPs, 0);
		if (bp_perm_counts != NULL) {
			numprocs = readNumberOfProcs(remoteFilepath, line, MAX_LINE_LENGTH, ops.rdir, ops.szRunId, SUFFIX_RBEPI_FILE); // The EPI file.
			if (numprocs > 0) {
				int i;
				BOOL hasCopiedFile = FALSE;
				char line[MAX_LINE_LENGTH];
				BOOL createdLocalFbepiFile = FALSE, createdLocalFepiFile = FALSE;

				// Copy the permutation RBEPI files to the local working directory.
				for (i = 0; i < numprocs; i++) {
					set_datafile_path(remoteFilepath, ops.rdir, ops.szRunId, i, SUFFIX_RBEPI_FILE);
					set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_RBEPI_FILE);
					if (copyFile(remoteFilepath, localFilepath)) hasCopiedFile = TRUE;
				}

				// If have copied a file, parse to get the permutation counts from each rank.
				if (hasCopiedFile) {
					for (i = 0; i < numprocs; i++) {
						FILE *in = NULL;

						set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_RBEPI_FILE);
						in = fopen(localFilepath, "r");
						if (in != NULL) {
							int snpIndex = 0;

							fgets(line, MAX_LINE_LENGTH, in); // Skip the numprocs header line.
							fgets(line, MAX_LINE_LENGTH, in); // The number of permutations
							nPermutations += atoi(line);

							// Loop through the rest of the file to get the permutation counts.
							while(fgets(line, MAX_LINE_LENGTH, in) != NULL) {
								bp_perm_counts[snpIndex] += atoi(line);
								snpIndex++;
							}
							fclose(in);
							deleteFile(localFilepath); // Remove the local file after parsing.
						}
					}

					// Open the BPEPI file to tag on the permuted counts to the output.
					sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_BEPI_FILE);
					sprintf(remoteFilepath, "%s%s%s", ops.rdir, ops.szRunId, SUFFIX_BEPI_FILE);
					if (copyFile(remoteFilepath, localFilepath)) {
						FILE *in = NULL, *out = NULL;

						sprintf(remoteFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_FBEPI_FILE); // Local output file path.

						in = fopen(localFilepath, "r");
						out = fopen(remoteFilepath, "w");
						if (in != NULL && out != NULL) {
							size_t line_length = 0;
							createdLocalFbepiFile = TRUE;

							// Read the original header line & add the 2 extra columns.
							fgets(line, MAX_LINE_LENGTH, in); 
							line_length = strlen(line);
							if (line_length > 0) {
								line[line_length-1] = ','; // chop off the new-line.
								line[line_length] = '\0';
								fprintf(out, "%sP,BP_PERM_RATIO\n", line);
							}
							while(fgets(line, MAX_LINE_LENGTH, in) != NULL) {
								if (strlen(line) > 0) {
									char *p, *q;
									char token[80], rs1[13], rs2[13], summary_sig[80], summary_good[80], best_score[80], best_snp[80];
									int tokenCount = 0, snpIndex1 = -1, snpIndex2 = -1;
									double bp_perm_ratio = 0, pvalue = 0; // Best pairwise permutation ratio.

									// Initialise the string, just incase.
									*rs1 = *rs2 = *summary_sig = *summary_good = *best_score = *best_snp = '\0';

									// Read a token at a time from the line
									p = line;
									while (*p) {
										q = token;
										while (*p != ',' && *p != '\r' && *p != '\n' && *p) {
											*q = *p;
											q++;
											p++;
										}
										if (*p) p++; // Next token
										*q = '\0'; // Null terminate token.

										switch (tokenCount) {
											case 0:
												snpIndex1 = atoi(token);
												break;
											case 1:
												strcpy(summary_sig, token);
												break;
											case 2:
												strcpy(summary_good, token);
												break;
											case 3:
												strcpy(best_score, token);
												break;
											case 4:
												snpIndex2 = atoi(token);
												strcpy(best_snp, token); // Need the string for the 'Na' error code.
												break;
											default:
												break;
										}
										tokenCount++;
									}

									// Get the RS numbers
									if (snpIndex1 >= 0 && snpIndex1 < nSNPs) { 
										strcpy(rs1, map[snpIndex1].name);

										// calculate the EMP1 value.
										bp_perm_ratio = (double) (bp_perm_counts[snpIndex1] + 1) / (double) (nPermutations + 1);
									}
									if (snpIndex2 >= 0 && snpIndex2 < nSNPs) {
										// If the pvalue set to 'Na', then this SNP has no best partner.
										if (strcmp(best_snp, "Na") == 0) bp_perm_ratio = 0;
										else {
											double score = atof(best_score);
											strcpy(rs2, map[snpIndex2].name);
											pvalue =  normdist(-sqrt(score)) * 2;
										}	
									}

									// Write out the line of output.
									fprintf(out, "%s,%s,%s,%s,%s,%10.16f,%5.16f\n", rs1, summary_sig, summary_good, best_score, rs2, pvalue, bp_perm_ratio);
								}
							}
							createdLocalFbepiFile = TRUE;
						}
						if (in != NULL) {
							fclose(in);
							sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_BEPI_FILE);
							deleteFile(localFilepath);
						}
						if (out != NULL) fclose(out);
					}
				}

				// Get the interesting interactions block.
				// Load the Interaction file containing interesting co-ordinates.
				sprintf(remoteFilepath, "%s%s%s", ops.rdir, ops.szRunId, SUFFIX_EPI_FILE2);
				interesting_interactions = int2_load(remoteFilepath, &nInterestingInteractions);
				if (interesting_interactions != NULL && nInterestingInteractions > 0) {
					qsort(interesting_interactions, nInterestingInteractions, sizeof(struct INT2), int2_cmp);

					// Copy the permutation REPI files to the local working directory.
					hasCopiedFile = FALSE;
					for (i = 0; i < numprocs; i++) {
						set_datafile_path(remoteFilepath, ops.rdir, ops.szRunId, i, SUFFIX_REPI_FILE);
						set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_REPI_FILE);
						if (copyFile(remoteFilepath, localFilepath)) hasCopiedFile = TRUE;
					}

					// Read the permutation results from each rank.
					if (hasCopiedFile) {
						FILE *out = NULL;

						for (i = 0; i < numprocs; i++) {
							FILE *in = NULL;

							set_datafile_path(localFilepath, ops.ldir, ops.szRunId, i, SUFFIX_REPI_FILE);

							in = fopen(localFilepath, "r");
							if (in != NULL) {
								struct INT2 interaction;

								fgets(line, MAX_LINE_LENGTH, in); // Skip the numprocs header line.
								fgets(line, MAX_LINE_LENGTH, in); // The number of permutations

								// Loop through the rest of the file to get the permutation counts.
								while(fgets(line, MAX_LINE_LENGTH, in) != NULL) {
									char token[80];
									int index = -1;

									// Read the interaction results from the output file.
									int2_init(&interaction);
									get_token(line, token, 0, ',');
									interaction.p1 = atoi(token);
									get_token(line, token, 1, ',');
									interaction.p2 = atoi(token);
									get_token(line, token, 2, ',');
									interaction.R = atoi(token);
									get_token(line, token, 3, ',');
									interaction.maxR = atoi(token);
									get_token(line, token, 4, ',');
									interaction.nPermutations = atoi(token);

									// Find the interaction in the interesting array.
									index = int2_search(interesting_interactions, nInterestingInteractions, &interaction);
									if (index >= 0) {
										interesting_interactions[index].R += interaction.R;
										interesting_interactions[index].maxR += interaction.maxR;
										interesting_interactions[index].nPermutations += interaction.nPermutations;
									}
								}
								fclose(in);
								deleteFile(localFilepath); // Remove the local file after parsing.
							}
						}

						// Write out the 'interesting interactions' results to the final output file.
						sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_FEPI_FILE);
						out = fopen(localFilepath, "w");
						if (out != NULL) {
							fprintf(out, "SNP1,SNP2,CHISQ,P,EMP1,EMP2\n");
							for (i = 0; i < nInterestingInteractions; i++) {
								char *name1 = map[interesting_interactions[i].p1].name, *name2 = map[interesting_interactions[i].p2].name;
								double chisq = interesting_interactions[i].z * interesting_interactions[i].z;
								double p = normdist(-sqrt(chisq)) * 2;
								double emp1 = (double) (interesting_interactions[i].R + 1) / (double) (interesting_interactions[i].nPermutations + 1);
								double emp2 = (double) (interesting_interactions[i].maxR + 1) / (double) (interesting_interactions[i].nPermutations + 1);

								fprintf(out, "%s,%s,%10.16f,%1.16f,%1.16f,%1.16f\n", name1, name2, chisq, p, emp1, emp2);
							}
							fclose(out);
							createdLocalFepiFile = TRUE; 
						}
					}
				}

				// Copy over the final output files & set the appropriate 'job done' flag.
				// If all data-writes successful, delete the temporary output files from the CFS.
				if (createdLocalFbepiFile) {
					BOOL createdRemoteFbepiFile = FALSE, createdRemoteFepiFile = FALSE;
					
					// Copy over the final 'Best Pair-wise file'.
					sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_FBEPI_FILE);
					sprintf(remoteFilepath, "%s%s%s", ops.rdir, ops.szRunId, SUFFIX_FBEPI_FILE);
					if (copyAndDeleteFile(localFilepath, remoteFilepath)) createdRemoteFbepiFile = TRUE;

					// Copy over the final 'interesting interactions' file.
					if (interesting_interactions != NULL) {
						sprintf(localFilepath, "%s%s%s", ops.ldir, ops.szRunId, SUFFIX_FEPI_FILE);
						sprintf(remoteFilepath, "%s%s%s", ops.rdir, ops.szRunId, SUFFIX_FEPI_FILE);
						if (sort_file(localFilepath, 5)) {
							deleteFile(localFilepath);
							strcat(localFilepath, SUFFIX_SORTED_FILE);
							if(copyAndDeleteFile(localFilepath, remoteFilepath)) createdRemoteFepiFile = TRUE;
						}
					}

					// Set the 'jobDone' flag, primary rule is no best-paire output, the run has failed.
					if (createdRemoteFbepiFile) {
						jobDone = TRUE;
						if (interesting_interactions != NULL && !createdRemoteFepiFile) jobDone = FALSE; // Failure of the 'interesting interactions' output.
					}

					// If the job is done, remove all the intermediate output files.
					if (jobDone) {
						// These temp files are always created on the central file server.
						sprintf(remoteFilepath, "%s%s%s", ops.rdir, ops.szRunId, SUFFIX_BEPI_FILE);
						deleteFile(remoteFilepath);
						sprintf(remoteFilepath, "%s%s%s", ops.rdir, ops.szRunId, SUFFIX_EPI_FILE);
						deleteFile(remoteFilepath);

						for (i = 0; i < numprocs; i++) {
							set_datafile_path(remoteFilepath, ops.rdir, ops.szRunId, i, SUFFIX_RBEPI_FILE);
							deleteFile(remoteFilepath);
						}

						// Remove the temporary 'interesting interactions' output files.
						if (interesting_interactions != NULL) {
							deleteFile(remoteFilepath);
							for (i = 0; i < numprocs; i++) {
								set_datafile_path(remoteFilepath, ops.rdir, ops.szRunId, i, SUFFIX_REPI_FILE);
								deleteFile(remoteFilepath);
							}
						}
					}
				}
			}
		} 
		if (bp_perm_counts != NULL) free(bp_perm_counts);
		if (perms_done != NULL) free(perms_done);
		if (interesting_interactions != NULL) free(interesting_interactions);
		if (map != NULL) free(map); 
	}
	
	if (jobDone) return 0;
	else return -1;
}

int main(int argc, char *argv[]) {
	return fast_pepi_merge_main(argc, argv);
}